use super::super::super::{
    common::{ErrorResult, RequestStruct},
    log, route,
    route::RouterResult,
    ProtocolTrait,
};
use super::MessageStruct;
use chrono::Local;
use std::fmt::Display;
use std::io::ErrorKind;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::mpsc::channel;

#[derive(Debug, Clone)]
pub struct Proto {
    ip: String,
}
impl Proto {
    pub fn new(ip: &str) -> Self {
        let ip = ip.to_string();
        Proto { ip }
    }
}
impl ProtocolTrait for Proto {
    async fn listen(&self) -> Result<(), ErrorResult> {
        let listener = TcpListener::bind(&self.ip).await;
        if let Err(e) = listener {
            return Err(ErrorResult {
                code: 2001,
                message: format!("Failed to bind to address: {:?}", e),
            });
        }

        log::info(format!("Listening on: {}", self.ip));
        let listener = listener.unwrap();
        'accept_loop: loop {
            // The second item contains the IP and port of the new connection.
            let tcp = listener.accept().await;
            if let Err(e) = tcp {
                log::error(format!("Failed to accept connection: {}", e));
                continue 'accept_loop;
            }
            let (tokio_tcp_stream, socket_addr) = tcp.unwrap();
            log::info(format!("{:?} Connection access by client", socket_addr));

            tokio::spawn(async move {
                let (read, mut write) = tokio_tcp_stream.into_split();
                let readable = read.readable().await;
                if let Err(e) = readable {
                    log::error(format!("Attempt to read data failed: {:?}", e));
                    let _ = write.shutdown();
                    return;
                }
                let access_time = Local::now().timestamp();
                // Create N:1 channels for writing information
                let (tx, mut rx) = channel::<Vec<u8>>(1024);
                tokio::spawn(async move {
                    while let Some(msg) = rx.recv().await {
                        let pack_result = pack(&msg);
                        if let PackStateEnum::ExceedingLengthLimit = pack_result.state {
                            log::error(format!(
                                "When writing packaging failed: {:?}",
                                pack_result.state
                            ));
                            continue;
                        }
                        let write_status = write.try_write(&pack_result.message);
                        if let Err(e) = write_status {
                            log::error(format!("{:?} When writing failed: {:?}", socket_addr, e));
                        }
                    }
                });

                let buf_max = 2048;
                let mut buffer = vec![0; buf_max];
                // Save messages for handling sticky and unpacking packages
                let mut message_vec = Vec::new();
                'read_loop: loop {
                    let try_read = read.try_read(&mut buffer);
                    if try_read.is_err() {
                        match try_read {
                            Err(e) if e.kind() == ErrorKind::WouldBlock => {
                                let readable = read.readable().await;
                                if readable.is_err() {
                                    return;
                                }
                                continue 'read_loop;
                            }
                            Err(e) => {
                                log::warn(format!("{:?} link Error: {:?}", socket_addr, e));
                                break;
                            }
                            _ => {}
                        }
                        break;
                    }
                    let read_length = try_read.unwrap();
                    if read_length == 0 {
                        log::warn(format!("{} Connection closed by client", socket_addr));
                        break;
                    }
                    message_vec.extend(buffer[0..read_length].to_vec());
                    while message_vec.len() != 0 {
                        if read_length < 5 {
                            // The message does not meet the requirements, clear message_cec
                            message_vec = Vec::new();
                            continue 'read_loop;
                        }

                        let parser_result = analysis(&mut message_vec);

                        // Data parsing failed, skip the current reception and clear the buffer
                        if let AnalysisStateEnum::DataIllegal = parser_result.state {
                            log::warn(format!("Error: {:?}", parser_result.state));
                            message_vec = Vec::new();
                            continue 'read_loop;
                        }

                        // The data has been unpacked and needs to continue receiving until the data is complete
                        if let AnalysisStateEnum::LengthInsufficient = parser_result.state {
                            message_vec = Vec::new();
                            continue 'read_loop;
                        }

                        // Data parsing failed, discard the current message and continue reading subsequent messages
                        if let AnalysisStateEnum::ParseFailed = parser_result.state {
                            message_vec = message_vec[parser_result.read_len..].to_vec();
                            continue 'read_loop;
                        }

                        // Execute middleware and perform route matching
                        let tx_clone = tx.clone();
                        let route_params = RequestStruct {
                            key: parser_result.route_key.clone(),
                            addr: socket_addr.to_string(),
                            access_time,
                            read: parser_result.message,
                            send: tx_clone,
                        };
                        let route_result =
                            route::call_route(&parser_result.route_key, route_params);

                        match route_result {
                            // Middleware blocked passage
                            RouterResult::MiddlewareBlocked => {
                                message_vec = message_vec[parser_result.read_len..].to_vec();
                                continue;
                            }
                            // Route not found
                            RouterResult::RouteMatchFail => {
                                message_vec = message_vec[parser_result.read_len..].to_vec();
                                continue;
                            }
                            // Route not initialized
                            RouterResult::RouteNotInitialized => {
                                break 'read_loop;
                            }
                            // Middleware blocks release and requests closure of current connection
                            RouterResult::MiddlewareBlockedAndClosed => {
                                break 'read_loop;
                            }
                            // Route execution completed
                            RouterResult::RouteExecuted => {}
                        };
                        // Determine whether there are unread sub packages
                        if let AnalysisStateEnum::UnreadSubPacket = parser_result.state {
                            // The data exists in a sub package and needs to continue processing
                            message_vec = message_vec[parser_result.read_len..].to_vec();
                        } else {
                            message_vec = Vec::new();
                        }
                    }
                    // After processing, clear the buffer
                    buffer.fill(0);
                }
            });
        }
    }
}

fn analysis(buffer: &Vec<u8>) -> AnalysisResult {
    let msg_len = get_message_len(buffer);
    let mut res = AnalysisResult::new();

    if msg_len == 0 {
        res.state = AnalysisStateEnum::DataIllegal;
        return res;
    }

    let len_position = 4;
    if msg_len > buffer.len() - len_position {
        res.state = AnalysisStateEnum::LengthInsufficient;
        return res;
    }
    // read data buffer
    // read length
    res.read_len = msg_len + len_position;
    let mut message = vec![0; msg_len];
    message.copy_from_slice(&buffer[len_position..res.read_len]);
    // get route_key
    let message_struct: MessageStruct = match serde_json::from_slice(&message) {
        Ok(message_struct) => message_struct,
        Err(_) => {
            res.state = AnalysisStateEnum::ParseFailed; // Message parsing failed
            return res;
        }
    };
    // Store messages
    res.message = message_struct.message.into_bytes();
    res.route_key = message_struct.route_key;
    // There are sub packages present
    if msg_len != buffer.len() - len_position {
        res.state = AnalysisStateEnum::UnreadSubPacket;
    }
    res
}

fn pack(buff: &Vec<u8>) -> PackResult {
    // get message length
    let mut res = PackResult::new();
    let data_length: u32 = match buff.len().try_into() {
        Ok(value) => value,
        Err(_) => {
            res.state = PackStateEnum::ExceedingLengthLimit;
            return res;
        }
    };
    res.write_len = data_length as usize + 4;
    let mut message = vec![0; 4];
    message.extend_from_slice(&buff);
    message[0..4].copy_from_slice(&data_length.to_be_bytes());
    res.message = message;
    res
}

#[derive(Debug)]
pub struct AnalysisResult {
    pub read_len: usize,
    pub route_key: String,
    pub message: Vec<u8>,
    pub state: AnalysisStateEnum,
}

impl AnalysisResult {
    pub fn new() -> Self {
        AnalysisResult {
            read_len: 0,
            route_key: String::new(),
            message: Vec::new(),
            state: AnalysisStateEnum::Success,
        }
    }
}

#[derive(Debug, Clone)]
pub enum AnalysisStateEnum {
    Success,
    LengthInsufficient,
    ParseFailed,
    UnreadSubPacket,
    DataIllegal,
}
impl Display for AnalysisStateEnum {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AnalysisStateEnum::Success => write!(f, "Parsing Success"),
            AnalysisStateEnum::LengthInsufficient => write!(f, "Message Length Insufficient"),
            AnalysisStateEnum::ParseFailed => write!(f, "Message Parsing Failed"),
            AnalysisStateEnum::UnreadSubPacket => write!(f, "Message Contains Unread Sub Packets"),
            AnalysisStateEnum::DataIllegal => write!(f, "The data format is illegal"),
        }
    }
}

#[derive(Debug)]
pub struct PackResult {
    pub write_len: usize,
    pub message: Vec<u8>,
    pub state: PackStateEnum,
}

impl PackResult {
    pub fn new() -> Self {
        PackResult {
            write_len: 0,
            message: Vec::new(),
            state: PackStateEnum::Success,
        }
    }
}

#[derive(Debug, Clone)]
pub enum PackStateEnum {
    Success,
    ExceedingLengthLimit,
}

impl Display for PackStateEnum {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            PackStateEnum::Success => write!(f, "Parsing Success"),
            PackStateEnum::ExceedingLengthLimit => {
                write!(f, "Message Exceeds Maximum Length Limit")
            }
        }
    }
}
/// get message length
/// u32 format data with a message length of 4 bytes in big endian sequence
fn get_message_len(message: &Vec<u8>) -> usize {
    let mut msg_len = 0;
    for (i, byte) in message.iter().enumerate().take(4) {
        msg_len += (*byte as usize) << ((3 - i) * 8);
    }
    msg_len
}
